package com.cl.spark.node;

import com.cl.spark.base.BaseSparkNode;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.enums.SparkNodeEnum;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.functions;
import org.springframework.stereotype.Component;

import java.util.Arrays;

@Component
public class MoreInputUnionNode extends BaseSparkNode {

    @Override
    public SparkResult process(SparkParam sparkParam) {

        Dataset<Row> dataset = sparkParam.getSparkResultList().stream().map(SparkResult::getDataset).filter(i -> i != null && !i.isEmpty()).reduce(Dataset::unionAll).orElse(null);

        return SparkResult.success(dataset);
    }

    @Override
    public SparkNodeEnum getType() {
        return SparkNodeEnum.MORE_INPUT_UNION;
    }
}
